package cn.hy.dataSync.transport.channel.disruptor;


import cn.hy.dataSync.config.CoreConstant;
import cn.hy.dataSync.config.TaskConfig;
import cn.hy.dataSync.element.Record;
import cn.hy.dataSync.transport.channel.Channel;
import cn.hy.dataSync.transport.record.TerminateRecord;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventTranslatorOneArg;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.Collection;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;

/**
 * @author tangmengyang
 */
public class DisruptorChannel extends Channel {

    private static final int MAXIMUM_CAPACITY = 1 << 16;

    private int bufferSize;

    private Disruptor<RecordEvent> disruptor;

    private RingBuffer<RecordEvent> ringBuffer;

    private RecordEventProducer eventProducer;

    private ArrayBlockingQueue<Record> queue = new ArrayBlockingQueue<>(this.getCapacity());

    public DisruptorChannel(TaskConfig taskConfig) {
        super(taskConfig);

        bufferSize = coreConfig.getInteger(CoreConstant.EXCHANGER_BUFFER_SIZE, 1);

        EventFactory<RecordEvent> eventFactory = new RecordEventFactory();
        int ringBufferSize = ringBufferSizeFor(this.getCapacity());
        disruptor = new Disruptor<>(
                eventFactory,
                ringBufferSize,
                Executors.defaultThreadFactory(),
                ProducerType.SINGLE,
                new YieldingWaitStrategy());

        disruptor.handleEventsWith(new RecordEventHandler(queue));
        disruptor.start();

        ringBuffer = disruptor.getRingBuffer();
        eventProducer = new RecordEventProducer(ringBuffer);
    }

    @Override
    protected void doPush(Record r) {
        long startTime = System.nanoTime();
        eventProducer.onData(r);
        waitWriterTime.addAndGet(System.nanoTime() - startTime);
    }

    @Override
    protected void doPushAll(Collection<Record> rs) {
        rs.forEach(this::doPush);
    }

    @Override
    protected Record doPull() {
        long startTime = System.nanoTime();
        try {
            Record record = this.queue.take();
            waitReaderTime.addAndGet(System.nanoTime() - startTime);
            return record;
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException(e);
        }
    }

    @Override
    protected void doPullAll(Collection<Record> rs) {
        assert rs != null;
        rs.clear();

        int okSize = 0;
        while (this.bufferSize > okSize) {
            rs.add(doPull());
            okSize++;
        }
    }

    @Override
    public int size() {
        return ringBuffer.getBufferSize();
    }

    @Override
    public boolean isEmpty() {
        return ringBuffer.hasAvailableCapacity(ringBuffer.getBufferSize());
    }

    @Override
    public void clear() {
    }

    @Override
    public void close() {
        super.close();
        this.doPush(TerminateRecord.get());
        this.disruptor.shutdown();
    }

    private static int ringBufferSizeFor(int c) {
        int n = c - 1;
        n |= n >>> 1;
        n |= n >>> 2;
        n |= n >>> 4;
        n |= n >>> 8;
        n |= n >>> 16;
        return (n < 0) ? 1 : (n >= MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : n + 1;
    }

    private static class RecordEventProducer {

        private static final EventTranslatorOneArg<RecordEvent, Record>
                TRANSLATOR = (event, sequence, record) -> event.setRecord(record);

        private final RingBuffer<RecordEvent> ringBuffer;

        RecordEventProducer(RingBuffer<RecordEvent> ringBuffer) {
            this.ringBuffer = ringBuffer;
        }

        void onData(Record record) {
            ringBuffer.publishEvent(TRANSLATOR, record);
        }
    }
}
